// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: daniel

#pragma once

#include <list>
#include <memory>
#include <queue>
#include <set>
#include <string>
#include <string_view>
#include <utility>

#include <boost/asio/io_context.hpp>
#include <boost/asio/post.hpp>
#include <glog/logging.h>

#include "coin/proto/coin_order_enums.pb.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/api_base/fix/fix_coin_logger.h"
#include "coin2/exchange/base/order/processor.h"
#include "quickfix/Application.h"
#include "quickfix/Initiator.h"
#include "quickfix/Message.h"
#include "quickfix/MessageStore.h"
#include "quickfix/SSLSocketInitiator.h"
#include "quickfix/SessionSettings.h"
#include "quickfix/SocketInitiator.h"

namespace coin2::exchange::base::api_base::fix {

class FixType {
 public:
  using RequestType = FIX::Message;
  using ResponseType = FIX::Message;
  using RequestTypePtr = std::unique_ptr<FIX::Message>;
  using SubscriberCB = std::function<void(const FIX::Message&)>;
  using Processor = coin2::exchange::base::order::BaseOrderProcessor;
  using RawMessage = coin2::exchange::base::order::BaseOrderProcessor::RawMessage;
  using OrderExecution = coin2::exchange::base::order::OrderExecution;
  using Executor = boost::asio::io_context;
};

/********************************************************************************
 * template class FixTradeClientBase                                            *
 ********************************************************************************/
template <class FixTradeClient>
class FixTradeClientBase {
 public:
  using RequestType = FixType::RequestType;
  using ResponseType = FixType::ResponseType;
  using RequestTypePtr = FixType::RequestTypePtr;
  using SubscriberCB = FixType::SubscriberCB;
  using Processor = FixType::Processor;
  using RawMessage = FixType::RawMessage;
  using OrderExecution = FixType::OrderExecution;
  using Executor = FixType::Executor;
  using SessionSettingsPtr = std::unique_ptr<FIX::SessionSettings>;
  using StoreFactoryPtr = std::unique_ptr<FIX::MemoryStoreFactory>;
  using LogFactoryPtr = std::unique_ptr<CoinLogFactory>;

 public:
  explicit FixTradeClientBase(std::string_view);
  virtual ~FixTradeClientBase();

  FixTradeClientBase(const FixTradeClientBase&) = delete;
  FixTradeClientBase& operator=(const FixTradeClientBase&) = delete;

  void StartBase();
  void StopBase();
  void Send(RequestTypePtr);
  void AsyncSend(RequestTypePtr);
  void AddSubscriberV2(SubscriberCB);

  inline auto GetExecutor() -> Executor* { return executor_; }
  inline void SetExecutor(Executor* executor) { executor_ = executor; }
  inline void SetProcessor(Processor* processor) { processor_ = processor; }
  inline bool IsAvailable() const { return IsInit() && IsLogon() && send_error_cnt_ == 0; }

 protected:
  void AssignSessionID(FIX::Message*);
  void AsyncSendImpl();
  void ProcessIncomeMsg(const FIX::Message&, const FIX::SessionID&);
  void PostSystemMsg(const coin::proto::ErrorCode& error_code);
  void AssignAuthKey(FIX::SessionSettings*);

  inline FIX::Initiator& getInitiator() {
    CHECK(initiator_);
    return *initiator_;
  }
  inline bool IsInit() const { return is_init_; }
  inline void IsInit(bool is_init) { is_init_ = is_init; }
  inline bool IsLogon() const { return is_logon_; }
  inline void IsLogon(bool is_logon) { is_logon_ = is_logon; }
  inline void IncreaseSendError() { send_error_cnt_++; }
  inline void DecreaseSendError() {
    if (send_error_cnt_ == 0) {
      return;
    }
    send_error_cnt_--;
  }
  inline std::string GetSessionSetting() const { return settings_file_; }
  inline void SetSessionSetting(std::string_view setting_file) { settings_file_ = setting_file; }
  inline std::string GetAuthKey() const { return auth_key_file_; }

 private:
  bool is_sending_ = false;  // support multiple threads?
  bool is_init_ = false;
  bool is_logon_ = false;
  int8_t send_error_cnt_ = 0;
  std::string auth_key_file_;
  std::string settings_file_;
  FixTradeClient* client_;
  SessionSettingsPtr settings_;
  StoreFactoryPtr store_factory_;
  LogFactoryPtr log_factory_;
  std::unique_ptr<FIX::Initiator> initiator_;
  std::queue<RequestTypePtr> requests_;
  std::list<SubscriberCB> subscribers_;
  Executor* executor_;
  Processor* processor_;
};  // template<class FixTradeClient> class FixTradeClientBase

template <class FixTradeClient>
FixTradeClientBase<FixTradeClient>::FixTradeClientBase(std::string_view auth_key_file)
    : auth_key_file_{auth_key_file},
      client_{nullptr},
      settings_{nullptr},
      store_factory_{nullptr},
      log_factory_{nullptr},
      initiator_{nullptr},
      executor_{nullptr},
      processor_{nullptr} {
  static_assert(
      std::is_base_of<FixTradeClientBase, FixTradeClient>::value,
      "FixTradeClientBase should be base class of FixTradeClient");
  client_ = static_cast<FixTradeClient*>(this);
}

template <class FixTradeClient>
FixTradeClientBase<FixTradeClient>::~FixTradeClientBase() {
  StopBase();
}

template <class FixTradeClient>
void FixTradeClientBase<FixTradeClient>::StartBase() {
  if (IsInit()) {
    return;
  }
  try {
    settings_ = std::make_unique<FIX::SessionSettings>(GetSessionSetting());
    AssignAuthKey(settings_.get());
    store_factory_ = std::make_unique<FIX::MemoryStoreFactory>();
    log_factory_ = std::make_unique<CoinLogFactory>(*settings_);
    bool use_non_ssl = false;
    if (settings_->get().has("UseNonSSL") && settings_->get().getBool("UseNonSSL")) {
      use_non_ssl = true;
    }
    LOG(INFO) << "Initiating " << (use_non_ssl ? "Non-SSL" : "SSL") << " fix client";
    if (!use_non_ssl) {
      initiator_ = std::make_unique<FIX::SSLSocketInitiator>(
          *client_,
          *store_factory_,
          *settings_,
          *log_factory_);
    } else {
      initiator_ = std::make_unique<FIX::SocketInitiator>(
          *client_,
          *store_factory_,
          *settings_,
          *log_factory_);
    }
    initiator_->start();
    IsInit(true);
  } catch (FIX::Exception& ex) {
    LOG(ERROR) << "exception: " << ex.what() << "\n";
    IsInit(false);
  } catch (std::exception& ex) {
    LOG(ERROR) << "exception: " << ex.what() << "\n";
    IsInit(false);
  }
}

template <class FixTradeClient>
void FixTradeClientBase<FixTradeClient>::StopBase() {
  IsInit(false);
  if (initiator_) {
    initiator_->stop();
  }
  // WARNING: initiator must be released before others
  initiator_.reset();
  log_factory_.reset();
  store_factory_.reset();
  settings_.reset();
}

template <class FixTradeClient>
void FixTradeClientBase<FixTradeClient>::AssignAuthKey(FIX::SessionSettings* session_setting) {
  client_->AssignSubSystemAuthKey(session_setting);
}

template <class FixTradeClient>
void FixTradeClientBase<FixTradeClient>::AssignSessionID(FIX::Message* message) {
  try {
    FIX::SessionID session_id = message->getSessionID();
    FIX::Session* session = FIX::Session::lookupSession(session_id);
    if (session != nullptr) {
      throw FIX::SessionNotFound(session_id.toString());
    }
    // session id is valid, return
    return;
  } catch (FIX::Exception& ex) {
    // do nothing
  }
  // session id is invalid, assign a session id
  const std::set<FIX::SessionID>& sessions = initiator_->getSessions();
  if (sessions.empty()) {
    throw FIX::ConfigError();
  }
  // TODO(daniel) - random assign session id
  message->setSessionID(*sessions.begin());
}

// Synchronization send
template <class FixTradeClient>
void FixTradeClientBase<FixTradeClient>::Send(RequestTypePtr message) {
  if (!IsAvailable()) {
    LOG(ERROR) << "fix client is unavailable\n";
    return;
  }
  try {
    AssignSessionID(message.get());
    FIX::SessionID session_id = message->getSessionID();
    FIX::Session* session = FIX::Session::lookupSession(session_id);
    if (session == nullptr) {
      throw FIX::SessionNotFound(session_id.toString());
    }
    session->send(*message);
  } catch (FIX::Exception& ex) {
    LOG(ERROR) << "exception: " << ex.what() << "\n";
  } catch (std::exception& ex) {
    LOG(ERROR) << "exception: " << ex.what() << "\n";
  }
}

// Asynchronization send
template <class FixTradeClient>
void FixTradeClientBase<FixTradeClient>::AsyncSend(RequestTypePtr message) {
  if (!IsAvailable()) {
    LOG(ERROR) << "fix client is unavailable\n";
    return;
  }
  if (!message) {
    LOG(ERROR) << "invalid message";
    return;
  }
  // TODO(daniel) - support multiple threads
  requests_.push(std::move(message));
  if (is_sending_) {
    return;
  }
  AsyncSendImpl();
}

template <class FixTradeClient>
void FixTradeClientBase<FixTradeClient>::AddSubscriberV2(SubscriberCB callback) {
  if (IsInit()) {
    LOG(ERROR) << "can't add subscriber after started initiator";
    return;
  }
  // TODO(daniel) - support multiple threads
  subscribers_.push_back(callback);
}

// Asynchronization send implementation
template <class FixTradeClient>
void FixTradeClientBase<FixTradeClient>::AsyncSendImpl() {
  is_sending_ = true;
  while (IsAvailable() && !requests_.empty()) {
    FIX::Message& message = *(requests_.front());
    try {
      AssignSessionID(message);
      FIX::SessionID session_id = message.getSessionID();
      FIX::Session* session = FIX::Session::lookupSession(session_id);
      if (session == nullptr) {
        throw FIX::SessionNotFound(session_id.toString());
      }
      if (!session->isLoggedOn()) {
        FIX::MsgType msg_type;
        message.getHeader().getFieldIfSet(msg_type);
        if (msg_type != "A" && msg_type != "5") {
          // TODO(daniel) - async wait
          sleep(1);
          continue;
        }
      }
      session->send(message);
    } catch (FIX::Exception& ex) {
      LOG(ERROR) << "exception on sending message: " << ex.what() << "\n(" << message << ")\n";
    } catch (std::exception& ex) {
      LOG(ERROR) << "exception on sending message: " << ex.what() << "\n(" << message << ")\n";
    }
    requests_.pop();
  }
  is_sending_ = false;
}

// process income message, call subscriber callback function
template <class FixTradeClient>
void FixTradeClientBase<FixTradeClient>::ProcessIncomeMsg(
    const FIX::Message& message,
    const FIX::SessionID& session_id) {
  // do the common part for every income message
  // call callback function
  for (auto& callback : subscribers_) {
    callback(message);
  }
  if (executor_ && processor_) {
    // DLOG(INFO) << "post parse fix message event: " << message;
    auto msg = std::make_unique<FIX::Message>(message);
    boost::asio::post(*executor_, [this, msg = std::move(msg)]() {
      RawMessage raw_msg;
      raw_msg.type = coin::proto::OrderSubTopicId::QUICKFIX_MESSAGE;
      raw_msg.fix_message = msg.get();
      raw_msg.timestamp = GetCurrentTimestamp();
      processor_->ProcessRawMessage(raw_msg);
    });
  }
}

template <class FixTradeClient>
void FixTradeClientBase<FixTradeClient>::PostSystemMsg(const coin::proto::ErrorCode& error_code) {
  if (executor_ && processor_) {
    boost::asio::post(*executor_, [this, error_code]() {
      RawMessage raw_msg;
      raw_msg.type = coin::proto::OrderSubTopicId::SYSTEM;
      raw_msg.error_code = error_code;
      raw_msg.timestamp = GetCurrentTimestamp();
      this->processor_->ProcessRawMessage(raw_msg);
    });
  }
}

}  // namespace coin2::exchange::base::api_base::fix
